package com.shujia.async

import java.util.concurrent.{CompletableFuture, ExecutorService, Executors}
import java.util.function.Consumer

import org.apache.flink.streaming.api.scala.async.{ResultFuture, RichAsyncFunction}
import com.shujia.bean._
import org.apache.flink.configuration.Configuration

class CarJoinKcAsync extends RichAsyncFunction[Car, CarWide] {
  var threadPoll: ExecutorService = _

  //open 每一个分区执行一次，在taskmanager中执行

  override def open(parameters: Configuration): Unit = {
    //创建线程池， 每一个分区一个
    threadPoll = Executors.newFixedThreadPool(8)
  }

  override def asyncInvoke(input: Car, resultFuture: ResultFuture[CarWide]): Unit = {
    // 设置客户端完成请求后要执行的回调函数
    // 回调函数只是简单地把结果发给 future
    CompletableFuture.supplyAsync(new MySupplier(input), threadPoll)
      //回调方法，get返回之后执行
      .thenAccept(new Consumer[CarWide] {

      //get完成之后回调方法
      override def accept(t: CarWide): Unit = {
        //返回结果
        resultFuture.complete(Iterable(t))
      }
    })
  }

  override def timeout(input: Car, resultFuture: ResultFuture[CarWide]): Unit = {
    println(" ==timeout==== " + input)
  }
}

